package com.shujia.flink.core;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Demo03Parallelism {
    public static void main(String[] args) throws Exception {
        /*
         * 如何设置并行度？
         * 1、考虑吞吐量
         *      有聚合操作的任务：1w条/s 一个并行度
         *      无聚合操作的任务：10w条/s 一个并行度
         * 2、考虑集群本身的资源
         *
         * Task的数量由并行度以及有无Shuffle一起决定
         *
         * Task Slot数量 是由任务中最大的并行度决定
         *
         * TaskManager的数量由配置文件中每个TaskManager设置的Slot数量及任务所需的Slot数量一起决定
         *
         */
        // FLink 并行度设置的集中方式
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 1、通过env设置，不推荐，如果需要台调整并行度得修改代码重新打包提交任务
//        env.setParallelism(3);

        DataStreamSource<String> ds = env.socketTextStream("master", 8888);

        // 2、每个算子可以单独设置并行度，视实际情况决定，一般不常用
        SingleOutputStreamOperator<Tuple2<String, Integer>> kvDS = ds
                        .map(word -> Tuple2.of(word, 1), Types.TUPLE(Types.STRING, Types.INT))
                        .setParallelism(4);


        SingleOutputStreamOperator<Tuple2<String, Integer>> wordCntDS2P =
                kvDS.keyBy(kv -> kv.f0)
                .sum(1)
                        .setParallelism(2);


        // 如果算子不设置并行度则以全局为准
        wordCntDS2P.print();

        // 3、还可以在提交任务的时候指定并行度，最常用 比较推荐的方式
        // 命令行：flink run 可以通过 -p 参数设置全局并行度
        // web UI：填写parallelism输入框即可设置，优先级：算子本身的设置 > env做的全局设置 > 提交任务时指定的 > 配置文件flink-conf.yaml

        env.execute();

    }
}
